package com.joven.shelltest.utils.redisdelay;

import com.joven.shelltest.dto.DealMessageDto;
import com.joven.shelltest.utils.RedisUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @Author wzj
 * @Date 2023-07-26 10:46
 **/
@Component
public class DelayedQueueProcessor {

    @Autowired
    RedisUtils redisUtils;

    private final Set<String> topic = new CopyOnWriteArraySet<>();

    public void addToQueue(String messageId, Object message, long delay) {
        redisUtils.zAdd(messageId, message, System.currentTimeMillis() + delay);
        topic.add(messageId);
    }

    /**
     * 每秒执行一次,轮询
     */
    @Scheduled(fixedDelay = 1000)
    public void processQueue() {
        for (String specialTopic : topic) {
//            String message = redisUtils.fetchOne(specialTopic);
            Object message = redisUtils.fetchOne(specialTopic);
//            if (StringUtils.isNotEmpty(message)) {
//                handleMessage(message);
//            }
            if (Objects.nonNull(message)) {
                handleMessage(message);
            }
        }

    }

    private void handleMessage(Object message) {
        // 处理消息的逻辑
//        System.out.println("处理消息：" + message + " at:" + LocalDateTime.now());
        DealMessageDto dealMessageDto = (DealMessageDto) message;
        System.out.println("处理消息：" + dealMessageDto.getMessage() + " at:" + LocalDateTime.now());
    }
}
